// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.query.filter.value;

import net.opentsdb.core.AppendDataPoints;
import net.opentsdb.core.Const;
import net.opentsdb.core.Internal;
import net.opentsdb.query.filter.TagVFilter;
import org.hbase.async.KeyValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public abstract class NumericExpression {

    private static final Logger log = LoggerFactory.getLogger(NumericExpression.class);
    public abstract boolean evaluate(Number theValue);

    public boolean evaluate(KeyValue keyValue){
        boolean flag;

        if(keyValue.qualifier().length % 2 == 0){

            Internal.Cell cell = Internal.parseSingleValue(keyValue);
            Number number = cell.parseValue();
            flag = evaluate(number);
        }else if(keyValue.qualifier()[0] == AppendDataPoints.APPEND_COLUMN_PREFIX){

            Number number;
            byte[] qualifier = keyValue.qualifier();

            int idx = 0;
            int qlength = 0;

            while (idx < keyValue.value().length) {
                qlength = Internal.getQualifierLength(keyValue.value(), idx);
                idx += qlength + Internal.getValueLengthFromQualifier(keyValue.value(), idx);
            }

            if (isInteger(qualifier)) {
                number = Internal.extractIntegerValue(keyValue.value(), idx-1,
                        (byte)Internal.getFlagsFromQualifier(qualifier));
            } else {
                number = Internal.extractFloatingPointValue(keyValue.value(), idx-1,
                        (byte)Internal.getFlagsFromQualifier(qualifier));
            }

            flag = evaluate(number);

            log.info("column idx {}, length {}, number {}", idx, qlength, number);
        }else{

            flag = false;
        }
        log.info("column length is {}, column : {}", keyValue.qualifier().length, keyValue);


        return flag;
    }

    public static NumericExpression fromTagVFilters(List<TagVFilter> filters){
        if( filters == null || filters.size() == 0 )
            return new IdentityExpr();

        AndExpr andExpr = new AndExpr();

        for (TagVFilter filter : filters) {
            if( filter.isFiltValue()){
                String type = filter.getType();
                if( type.equalsIgnoreCase(TagVEqualFilter.FILTER_NAME)){
                    andExpr.addExpression(new Equal(Double.parseDouble( filter.getFilter() )));
                }else if( type.equalsIgnoreCase(TagVNotEqualFilter.FILTER_NAME)){
                    andExpr.addExpression(new NotEqual(Double.parseDouble( filter.getFilter() )));
                }else if( type.equalsIgnoreCase(TagVGreatFilter.FILTER_NAME)){
                    andExpr.addExpression(new GreateThan(Double.parseDouble( filter.getFilter() )));
                }else if( type.equalsIgnoreCase(TagVGreatEqualFilter.FILTER_NAME)){
                    andExpr.addExpression(new GreateEqual(Double.parseDouble( filter.getFilter() )));
                }else if( type.equalsIgnoreCase(TagVLessEqualFilter.FILTER_NAME)){
                    andExpr.addExpression(new LessEqual(Double.parseDouble( filter.getFilter() )));
                }else if( type.equalsIgnoreCase(TagVLessFilter.FILTER_NAME)){
                    andExpr.addExpression(new LessThan(Double.parseDouble( filter.getFilter() )));
                }else if( type.equalsIgnoreCase(TagVBetweenFilter.FILTER_NAME)){
                    String filterstr = filter.getFilter();
                    String[] values = filterstr.split("~");
                    if( values.length!=2){
                        throw new IllegalArgumentException("ill between form:"+filterstr+",should be bt(number～number)");
                    }
                    andExpr.addExpression(new Between(Double.parseDouble( values[0] ),Double.parseDouble( values[1] )));
                } else{
                    throw new IllegalArgumentException("unknown value filter:"+type);
                }
            }
        }
        return andExpr;
    }

    public boolean isInteger(byte[] qualifier) {
        return (Internal.getFlagsFromQualifier(qualifier) &
                Const.FLAG_FLOAT) == 0x0;
    }
}
